// Created: Dec 14, 2014
//
// Author: wjinwen.1988@gmail.com

#include "chi/net/tcpconnection.h"

#include <errno.h>
#include <stdio.h>
#include <unistd.h>

#include <functional>

#include <glog/logging.h>

#include "chi/net/event_loop.h"
#include "chi/net/channel.h"
#include "chi/net/socket.h"
#include "chi/net/buffer.h"

namespace chi {
namespace net {

void DefaultConnectionCallback(const TcpConnectionPtr& conn) {
  LOG(INFO) << conn->peer_addr().ToIpPort() << " -> " 
            << conn->local_addr().ToIpPort() << " is " 
            << (conn->connected() ? "UP" : "DOWN");
}

void DefaultMessageCallback(const TcpConnectionPtr& conn, 
                            Buffer* message) {
  VLOG(11) << "receive " << message->ReadableBytes() << " bytes";
  message->RetrieveAll();
}
    
TcpConnection::TcpConnection(EventLoop* loop,
                             const std::string& name,
                             int sockfd,
                             const InetAddress& local_addr,
                             const InetAddress& peer_addr)
  : loop_(loop),
    name_(name),
    state_(StateE::kConnecting),
    socket_(new Socket(sockfd)),
    channel_(new Channel(loop, sockfd)),
    local_addr_(local_addr),
    peer_addr_(peer_addr) {
  channel_->SetReadCallback(std::bind(&TcpConnection::HandleRead, this));
  channel_->SetWriteCallback(std::bind(&TcpConnection::HandleWrite, this));
  VLOG(11) << "TcpConnection::ctor[" << name_ << "] at fd=" << sockfd;
  socket_->SetKeepAlive(true);
  socket_->SetSoBlocking(false);  // set nonblocking
}

TcpConnection::~TcpConnection() {
  VLOG(11) << "TcpConnection::dtor[" << name_ << "] at fd="  << channel_->fd();

  if (channel_) 
    delete channel_;
  if (socket_) 
    delete socket_;
}

void TcpConnection::Send(const std::string& message) {
  Send(message.data(), message.size());
}

void TcpConnection::Send(const void* data, size_t len) {
  if (state_ == StateE::kConnected) {
    std::string message(static_cast<const char*>(data), len);
    loop_->RunInLoop(std::bind(&TcpConnection::SendInLoop, this, message));
  }
}

void TcpConnection::Send(Buffer&& message) {
  Send(message.Peek(), message.ReadableBytes());
}

void TcpConnection::Shutdown() {
  if (state_ == StateE::kConnected) {
    set_state(StateE::kDisconnecting);
    loop_->RunInLoop(std::bind(&TcpConnection::ShutdownInLoop, this));
  }
}

void TcpConnection::SetTcpNoDelay(bool on) {
  socket_->SetTcpNoDelay(on);
}

void TcpConnection::set_connection_callback(const ConnectionCallback& cb) {
  connection_callback_ = cb;
}

void TcpConnection::set_message_callback(const MessageCallback& cb) {
  message_callback_ = cb;
}

void TcpConnection::set_close_callback(const CloseCallback& cb) {
  close_callback_ = cb;
}

void TcpConnection::ConnectionEstablished() {
  set_state(kConnected);
  channel_->EnableReading();
  connection_callback_(shared_from_this());
}

void TcpConnection::ConnectDestroyed() {
  if (state_ == StateE::kConnected) {
    set_state(kDisconnected);
    channel_->DisableAll();
  }
  
  connection_callback_(shared_from_this());
}

void TcpConnection::HandleRead() {
  int saved_errno = 0;
  ssize_t n = input_buffer_.ReadFd(channel_->fd(), &saved_errno);
  if (n > 0) {
    message_callback_(shared_from_this(), &input_buffer_);
  } else if (n == 0) {
    HandleClose();
  } else {
    errno = saved_errno;
    LOG(ERROR) << "read error " << errno;
    HandleError();
  }
}

void TcpConnection::HandleWrite() {
  if (channel_->IsWriting()) {
    ssize_t n = ::write(channel_->fd(),
                        output_buffer_.Peek(),
                        output_buffer_.ReadableBytes());
    if (n > 0) {
      output_buffer_.Retrieve(n);
      if (output_buffer_.ReadableBytes() == 0) {
        channel_->DisableWriting();
        if (state_ == kDisconnecting) {
          ShutdownInLoop();
        }
      } else {
        VLOG(11) << "I am going to write more data";
      }
    } else {
      LOG(ERROR) << "wirte error";
    } // end if (n > 0)
  } else {
    VLOG(11) << "connection fd=" << channel_->fd() << " is down, no more writing";
  } // end if (channel_->IsWriting())
}

void TcpConnection::HandleClose() {
  VLOG(11) << "fd = " << channel_->fd() << " state = " << state_;
  assert(state_ == StateE::kConnected || state_ == StateE::kDisconnecting);
  set_state(StateE::kDisconnected);
  channel_->DisableAll();

  connection_callback_(shared_from_this());
  close_callback_(shared_from_this());
}

void TcpConnection::HandleError() {
  LOG(ERROR) << "socket error";
}

void TcpConnection::SendInLoop(const std::string& message) {
  RealSend(message.data(), message.size());
}

void TcpConnection::RealSend(const void* data, size_t len) {
  ssize_t nwrote = 0;
  ssize_t remaining = len;
  bool error = false;
  if (state_ == StateE::kDisconnected) {
    LOG(WARNING) << "disconnected, give up writing";
  }
  // if nothing in output queue, try writing directly
  if (!channel_->IsWriting() && output_buffer_.ReadableBytes() == 0) {
    nwrote = ::write(channel_->fd(), data, len);
    if (nwrote >= 0) {
      remaining = len - nwrote;
    } else { // nwrote < 0
      nwrote = 0;
      if (errno != EWOULDBLOCK) {
        LOG(ERROR) << "send error";
        if (errno == EPIPE) {
          error = true;
        }
      }
    }
  }

  assert((size_t)remaining <= len);
  if (!error && remaining > 0) {
    VLOG(11) << "I am going to write more data";
    output_buffer_.Append(static_cast<const char*>(data)+nwrote, remaining);
    if (!channel_->IsWriting()) {
      channel_->EnableWriting();
    }
  }
}

void TcpConnection::ShutdownInLoop() {
  if (!channel_->IsWriting()) {
    // we are not writing
    socket_->ShutdownWrite();
  }
}

} 
}

                             
